<?php

namespace app\common\library\queuemqtt;

use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\Exceptions\MqttClientException;
use think\Config;
use think\Log;

/**
 * 给队列使用的Mqtt发布消息类
 * Class QueueMqttPublisher
 * @package app\common\queuemqtt
 */
class QueueMqttPublisher
{
    private $mqtt;

    public function __construct()
    {
        $config = Config::get('mqtt');
        $this->mqtt = new MqttClient($config['server'], $config['port'], $config['client_id']);
        $connectionSettings = (new ConnectionSettings)
            ->setConnectTimeout(60)     // 连接超时时间
            ->setSocketTimeout(60)      // Socket超时时间
            ->setResendTimeout(10)      // 重发超时时间
            ->setUsername($config['username'])
            ->setPassword($config['password'])
            ->setKeepAliveInterval(60);// Keep Alive 间隔
        try {
            $this->mqtt->connect($connectionSettings, true);
            Log::error('Queue Connected to MQTT broker.');
        } catch (MqttClientException $e) {
            Log::error('Queue Mqtt连接失败: ' . $e->getMessage());
            throw $e;
        }
    }

    /**
     * 发布消息
     * @param $topic
     * @param $data
     * @throws MqttClientException
     */
    public function publishMessage($topic, $data)
    {
        try {
            Log::error("队列发送数据，topic:{$topic} ,data:".json_encode($data));
            $this->mqtt->publish($topic, json_encode($data,JSON_UNESCAPED_SLASHES|JSON_UNESCAPED_UNICODE), 0);
        } catch (MqttClientException|\Exception $e) {
            Log::error('Queue MQTT publish error: ' . $e->getMessage());
            throw $e;
        }
    }


    public function __destruct()
    {
        if ($this->mqtt !== null) {
            $this->mqtt->disconnect();
            Log::error('Queue Disconnected from MQTT broker.');
        }
    }
}
